package zero.tech.redis.topic.config;

import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import com.alibaba.fastjson.JSONObject;

import zero.tech.core.util.EngineLogUtil;
import zero.tech.core.work.WorkManager;
import zero.tech.redis.topic.RedisTopicProcess;
import zero.tech.redis.topic.TopicManager;

public class RedisMessageListener implements MessageListener {
	private RedisTemplate<String, Object> redisTemplate;
	private TopicManager topicManager;

	public RedisMessageListener(RedisTemplate<String, Object> redisTemplate, TopicManager topicManager) {
		this.redisTemplate = redisTemplate;
		this.topicManager = topicManager;
	}

	@SuppressWarnings("rawtypes")
	@Override
	public void onMessage(Message message, byte[] pattern) {
		try {
			StringRedisSerializer stringRedisSerializer = (StringRedisSerializer) redisTemplate.getKeySerializer();
			Jackson2JsonRedisSerializer jackson2JsonRedisSerializer=(Jackson2JsonRedisSerializer) redisTemplate.getValueSerializer();
			String channel = stringRedisSerializer.deserialize(message.getChannel());
			JSONObject jsonObject = (JSONObject) jackson2JsonRedisSerializer.deserialize(message.getBody());
			String redisOrder = "default";
			if (jsonObject.containsKey("redis_order")) {
				redisOrder = jsonObject.getString("redis_order");
			}
			String paramString = null;
			if (jsonObject.containsKey("param_string")) {
				paramString = jsonObject.getString("param_string");
			}
			RedisTopicProcess process = topicManager.getRedisMqOrderProcess(redisOrder, channel);
			if (process != null) {
				try {
					if (StringUtils.isNotBlank(paramString)) {
						process.init(paramString);
					}
					WorkManager.getManager().submit(process);
				} catch (Exception e) {
					EngineLogUtil.traceFull(e);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		} 

	}

}
